package com.xtayfjpk.zookeeper.test.master_workers;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

import com.xtayfjpk.zookeeper.test.master_workers.callback.client.CreateTaskCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.client.DeleteTaskResultCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.client.GetTaskResultCallback;
import com.xtayfjpk.zookeeper.test.master_workers.callback.client.TaskExistsCallback;
import com.xtayfjpk.zookeeper.test.master_workers.watcher.TaskStatusWatcher;

public class Client implements Watcher {
	private static final String DEFAULT_HOST_PORT = "192.168.0.202:2181";
	private static final Logger LOGGER = Logger.getLogger(Client.class);
	private ZooKeeper zk;
	private String hostPort;
	private volatile boolean connected = false;
    private volatile boolean expired = false;
    
    private ConcurrentHashMap<String, Object> ctxMap = new ConcurrentHashMap<>();
	
	public Client(String hostPort) {
		this.hostPort = hostPort;
	}
	
	public void startZK() throws IOException {
		zk = new ZooKeeper(hostPort, 15000, this);
	}
	
	@Override
	public void process(WatchedEvent event) {
		LOGGER.info(event);
		if(event.getType() == EventType.None) {
			switch (event.getState()) {
			case SyncConnected:
				connected = true;
				break;
			case Disconnected:
				connected = false;
				break;
			case Expired:
				expired = true;
				connected = false;
				LOGGER.info("Exiting due to session expiration");
				break;
			default:
				break;
			}
		}
	}
	
	public void sumbitTask(TaskObject task) {
		zk.create(
				"/tasks/task-", 
				task.getTask().getBytes(), 
				Ids.OPEN_ACL_UNSAFE, 
				CreateMode.PERSISTENT_SEQUENTIAL, 
				new CreateTaskCallback(this), 
				task);
	}
	
	
	public void watchTaskStatus(String path, Object ctx) {
		ctxMap.put(path, ctx);
		zk.exists(
				path, 
				new TaskStatusWatcher(this), 
				new TaskExistsCallback(this), 
				ctx);
	}
	
	
	public void getTaskResult(String path) {
		zk.getData(
				path, 
				false, 
				new GetTaskResultCallback(this), 
				ctxMap.get(path));
		
	}
	
	public void deleteTaskResult(String path) {
		zk.delete(
				path, 
				-1, 
				new DeleteTaskResultCallback(this), 
				null);		
	}
	
	public boolean isConnected() {
		return connected;
	}
	
	public boolean isExpired() {
		return expired;
	}
	
	public ConcurrentHashMap<String, Object> getCtxMap() {
		return ctxMap;
	}
	
    public static void main(String args[]) throws Exception {
        Client c = new Client(DEFAULT_HOST_PORT);
        c.startZK();
        
        while(!c.isConnected()) {
        	Thread.sleep(100);
        }
        
        TaskObject task1 = new TaskObject("Sample task", "sampleTask");
        TaskObject task2 = new TaskObject("Another sample task", "anotherSampleTask");
        
        c.sumbitTask(task1);
        c.sumbitTask(task2);
        
        task1.waitUntilDone();
        task2.waitUntilDone();
        
    }

	

	
}
